package com.shxzz.easy12306.framework.starter.idempotent.core.handler.spel;

import com.shxzz.easy12306.framework.starter.cache.DistributeCache;
import com.shxzz.easy12306.framework.starter.idempotent.annotation.Idempotent;
import com.shxzz.easy12306.framework.starter.idempotent.core.IdempotentAspect;
import com.shxzz.easy12306.framework.starter.idempotent.core.IdempotentContext;
import com.shxzz.easy12306.framework.starter.idempotent.core.exception.MessageRepeatConsumptionException;
import com.shxzz.easy12306.framework.starter.idempotent.core.handler.AbstractIdempotentExecuteHandler;
import com.shxzz.easy12306.framework.starter.idempotent.core.handler.IdempotentParamWrapper;
import com.shxzz.easy12306.framework.starter.idempotent.enums.MessageConsumerStatus;
import com.shxzz.easy12306.framework.starter.idempotent.toolkit.SpELUtil;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.concurrent.TimeUnit;

/**
 * 标识 key 类型是 SpEL 表达式的幂等执行处理器，适用于 MQ 场景
 * <p>
 *     消息执行逻辑：
 *     <ol>
 *         <li>第一次接收到消息时，执行方法消费，标记状态消费中</li>
 *         <li>重复接收到消息时，执行幂等操作
 *             <ul>
 *                 <li>消息正在消费，抛异常，等待消息重试验证消费正确性</li>
 *                 <li>消息已消费成功，不抛异常，消息确认</li>
 *             </ul>
 *         </li>
 *         <li>消费成功，标记状态已消费成功，消息确认</li>
 *         <li>消费异常，抛异常，等待消息重试</li>
 *
 *     </ol>
 * </p>
 * @author shxzz
 * @create 2024-02-23-18:55
 */
@RequiredArgsConstructor
public class IdempotentSpELByMQExecuteHandler extends AbstractIdempotentExecuteHandler implements IdempotentSpELService {

    private final DistributeCache distributeCache;

    private static final long CONSUMING_STATUS_TIMEOUT = 600L;
    private static final String WRAPPER = "param:SpEL:MQ";

    @SneakyThrows
    @Override
    protected IdempotentParamWrapper buildIdempotentParam(ProceedingJoinPoint proceedingJoinPoint) {
        Idempotent idempotent = IdempotentAspect.getIdempotent(proceedingJoinPoint);
        String lockKey = (String) SpELUtil.parseKey(idempotent.spel(), ((MethodSignature) proceedingJoinPoint.getSignature()).getMethod(), proceedingJoinPoint.getArgs());
        return IdempotentParamWrapper.builder()
                .lockKey(lockKey)
                .idempotent(idempotent)
                .proceedingJoinPoint(proceedingJoinPoint)
                .build();
    }

    @Override
    protected void handler(IdempotentParamWrapper idempotentParamWrapper) {
        String uniqueKey = idempotentParamWrapper.getIdempotent().uniqueKeyPrefix() + idempotentParamWrapper.getLockKey();

        StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) distributeCache.getInstance();
        Boolean isAbsent = stringRedisTemplate.opsForValue()
                .setIfAbsent(uniqueKey, MessageConsumerStatus.CONSUMING.getCode(),CONSUMING_STATUS_TIMEOUT, TimeUnit.SECONDS);
        if(Boolean.FALSE.equals(isAbsent)) {
            String msgStatusCode = stringRedisTemplate.opsForValue().get(idempotentParamWrapper.getLockKey());
            throw new MessageRepeatConsumptionException(MessageConsumerStatus.isConsuming(msgStatusCode), uniqueKey);
        }
        IdempotentContext.put(WRAPPER, idempotentParamWrapper);
    }

    @Override
    public void postProcessing() {
        IdempotentParamWrapper idempotentParamWrapper = (IdempotentParamWrapper) IdempotentContext.getValue(WRAPPER);
        if(idempotentParamWrapper == null) return;
        String uniqueKey = idempotentParamWrapper.getIdempotent().uniqueKeyPrefix() + idempotentParamWrapper.getLockKey();
        distributeCache.put(uniqueKey, MessageConsumerStatus.CONSUMED.getCode(), idempotentParamWrapper.getIdempotent().keyTimout(), TimeUnit.SECONDS);
    }

    @Override
    public void exceptionProcessing() {
        IdempotentParamWrapper idempotentParamWrapper = (IdempotentParamWrapper) IdempotentContext.getValue(WRAPPER);
        if(idempotentParamWrapper == null) return;

        String uniqueKey = idempotentParamWrapper.getIdempotent().uniqueKeyPrefix() + idempotentParamWrapper.getLockKey();
        distributeCache.delete(uniqueKey);
    }
}
